package cn.me.alphamq.broker.util;

import cn.me.alphamq.broker.constant.BrokerConstant;
import cn.me.alphamq.broker.model.consumequeue.ConsumeQueueIndexModel;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.lang.reflect.Method;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.security.AccessController;
import java.security.PrivilegedAction;

/**
 * 内存映射的工具类，支持基于 Java 的 MMap api 访问文件能力（文件的读写）
 * <p>
 * 支持指定的 offset 的文件映射（映射的内存体积 = 结束映射的 offset - 开始映射的 offset）
 * <p>
 * 文件从指定的 offset 开始读取
 * <p>
 * 文件从指定的 offset 开始写入
 * <p>
 * 文件映射后的内存释放
 *
 * @author f
 */
public class MMapUtil {

    private File file;
    private MappedByteBuffer mappedByteBuffer;
    private FileChannel fileChannel;

    public static void main(String[] args) throws IOException {
//        MMapUtil commitLog = new MMapUtil();
//        commitLog.loadFileInMMap("/Users/f/Program/Idea/AlphaMQ/broker/commitlog/test_topic/00000000",
//                0, BrokerConstant.COMMIT_LOG_MAX_SIZE);
//        System.out.println(new String(commitLog.readContent(0, 1000)));

        MMapUtil consumeQueue = new MMapUtil();
        consumeQueue.loadFileInMMap("/Users/f/Program/Idea/AlphaMQ/broker/consumequeue/test_topic/1/00000000",
                0, BrokerConstant.CONSUME_QUEUE_MAX_SIZE);
        int offset = 0;
        while (true) {
            byte[] content = consumeQueue.readContent(offset, 12);
            ConsumeQueueIndexModel model = new ConsumeQueueIndexModel(content);
            if (model.getMsgLength() == 0) {
                break;
            }
            System.out.println(model);
            offset += 12;
        }
    }

    /**
     * 指定 offset 做文件的映射
     *
     * @param filePath    文件路径
     * @param startOffset 开始映射的 offset
     * @param mappedSize  映射的体积（单位：byte）
     */
    public void loadFileInMMap(String filePath, int startOffset, int mappedSize) throws IOException {
        file = new File(filePath);
        if (!file.exists()) {
            throw new FileNotFoundException("filePath is " + filePath + " invalid");
        }

        fileChannel = new RandomAccessFile(file, "rw").getChannel();
        mappedByteBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, startOffset, mappedSize);
    }

    /**
     * 从文件的指定 offset 开始读取内容
     *
     * @param readOffset
     * @param size
     * @return
     */
    public byte[] readContent(int readOffset, int size) {
        ByteBuffer readBuffer = this.mappedByteBuffer.slice();
        readBuffer.position(readOffset);
        byte[] content = new byte[size];
        readBuffer.get(content);
        return content;
//        mappedByteBuffer.position(readOffset);
//        byte[] content = new byte[size];
//        for (int i = 0; i < size; i++) {
//            // 这里是从内存空间读取数据
//            content[i] = mappedByteBuffer.get(readOffset + i);
//        }
//        return content;
    }

    /**
     * 更高性能的写入 api，异步刷盘
     *
     * @param content
     */
    public void writeContent(byte[] content) {
        writeContent(content, false);
    }

    /**
     * 写入数据到磁盘当中，可以选择同步还是异步刷盘
     *
     * @param content
     * @param force
     */
    public void writeContent(byte[] content, boolean force) {
        // 默认刷到 page cache 中，如果需要强制刷盘，这里需要兼容
        mappedByteBuffer.put(content);
        if (force) {
            // 强制刷盘
            mappedByteBuffer.force();
        }
    }

    /**
     * 文件映射后的内存释放
     */
    public void clean() {
        if (mappedByteBuffer == null || !mappedByteBuffer.isDirect() || mappedByteBuffer.capacity() == 0) {
            return;
        }
        // 安全地（有权限的？）调用 DirectByteBuffer 对象中的 cleaner 字段的 clean() 方法
        invoke(invoke(viewed(mappedByteBuffer), "cleaner"), "clean");
    }

    // 通过 DirectByteBuffer 对象中的 att 字段，来获取最终的 DirectByteBuffer
    private ByteBuffer viewed(ByteBuffer buffer) {
        String methodName = "viewedBuffer";
        Method[] methods = buffer.getClass().getMethods();
        for (Method method : methods) {
            if ("attachment".equals(method.getName())) {
                methodName = "attachment";
                break;
            }
        }

        // 如果 ByteBuffer 的 att 字段没有值，则说明这个 ByteBuffer 不是 DirectByteBuffer
        // 它的 att 字段指向的 ByteBuffer 可能是 DirectByteBuffer
        ByteBuffer viewedBuffer = (ByteBuffer) invoke(buffer, methodName);
        if (viewedBuffer == null) {
            return buffer;
        } else {
            return viewed(viewedBuffer);
        }
    }

    // 在安全的环境下调用指定方法
    private Object invoke(final Object target, final String methodName, final Class<?>... args) {
        return AccessController.doPrivileged(new PrivilegedAction<Object>() {
            @Override
            public Object run() {
                try {
                    Method method = method(target, methodName, args);
                    method.setAccessible(true);
                    return method.invoke(target);
                } catch (Exception e) {
                    throw new IllegalStateException(e);
                }
            }
        });
    }

    // 获取 指定对象拥有的指定名称和参数 的方法
    private Method method(Object target, String methodName, Class<?>... args) throws NoSuchMethodException {
        try {
            return target.getClass().getMethod(methodName, args);
        } catch (NoSuchMethodException e) {
            return target.getClass().getDeclaredMethod(methodName, args);
        }
    }

}
